package dongshi.daddy.zookeeper.lock;

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 基于Ephemera 节点实现分布式锁测试类
 */
public class EphemeraNodeLockTest {
    private static final String SERVER_HOST = "127.0.0.1:5181";
    private static final int SESSION_TIME_OUT = 2000;
    private static ZooKeeper zooKeeper1;
    private static ZooKeeper zooKeeper2;
    private static ZooKeeper zooKeeper3;
    private static final String LOCK_KEY_NAME = "/lock";

    // 初始化客户端连接类
    static {
        try {
            zooKeeper1 = new ZooKeeper(SERVER_HOST, SESSION_TIME_OUT, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    //获取事件的状态
                    Event.KeeperState state = watchedEvent.getState();
                    //判断是否是连接事件
                    if (Event.KeeperState.SyncConnected == state) {
                        Event.EventType type = watchedEvent.getType();
                        if (Event.EventType.None == type) {
                            System.out.println("zk1客户端已连接...");
                        }
                    }
                }
            });
            zooKeeper2 = new ZooKeeper(SERVER_HOST, SESSION_TIME_OUT, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    //获取事件的状态
                    Event.KeeperState state = watchedEvent.getState();
                    //判断是否是连接事件
                    if (Event.KeeperState.SyncConnected == state) {
                        Event.EventType type = watchedEvent.getType();
                        if (Event.EventType.None == type) {
                            System.out.println("zk2客户端已连接...");
                        }
                    }
                }
            });
            zooKeeper3 = new ZooKeeper(SERVER_HOST, SESSION_TIME_OUT, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    //获取事件的状态
                    Event.KeeperState state = watchedEvent.getState();
                    //判断是否是连接事件
                    if (Event.KeeperState.SyncConnected == state) {
                        Event.EventType type = watchedEvent.getType();
                        if (Event.EventType.None == type) {
                            System.out.println("zk3客户端已连接...");
                        }
                    }
                }
            });
        } catch (
    IOException e) {
            e.printStackTrace();
        }
    }

    // 模拟三个服务上的客户端
    private static ZooKeeper[] zkArr = new ZooKeeper[] { zooKeeper1, zooKeeper2, zooKeeper3 };
    static CountDownLatch countDownLatch = new CountDownLatch(zkArr.length);

    public static void main(String[] args) throws Exception {
        // 模拟多个客户单争抢
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < zkArr.length; i++) {
            int finalI = i;
            executorService.submit(new Thread(() -> tryAcquireLock(zkArr[finalI]), "线程-" + i));
        }

        countDownLatch.await();
        executorService.shutdown();
    }

    private static void tryAcquireLock(ZooKeeper zooKeeper) {
        boolean canClose = false;
        try {
            zooKeeper.create(LOCK_KEY_NAME , "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.err.println(Thread.currentThread().getName() + " 争抢锁成功，开始执行临界区业务！");
            Thread.sleep((long) (Math.random() * 5000));
            System.out.println(Thread.currentThread().getName() + " 执行临界区业务结束！");
            // 业务执行完毕，可以关闭
            canClose = true;
            countDownLatch.countDown();
        } catch (KeeperException.NodeExistsException e) {
            try {
                zooKeeper.exists(LOCK_KEY_NAME, watchedEvent -> {
                    // 删除
                    if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted) {
                        System.out.println("锁被释放了，开始抢！");
                        try {
                            tryAcquireLock(zooKeeper);
                        } catch (Exception interruptedException) {
                            interruptedException.printStackTrace();
                        }
                    }
                });
                System.out.println(Thread.currentThread().getName() + " 争抢锁失败，注册监听下次再抢！！！");
            } catch (KeeperException keeperException) {
                keeperException.printStackTrace();
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
        } catch (Exception e) {
        } finally {
            try {
                if (canClose) zooKeeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
